package com.jy.oms.framework.asyn.task.executor;

import com.jy.oms.framework.asyn.task.collector.TimeoutTaskCollector;
import com.jy.oms.framework.asyn.task.constants.TaskConstants;
import com.jy.oms.framework.asyn.task.constants.TaskQueueTypeEnum;
import com.jy.oms.framework.asyn.task.core.ExecutorContext;
import com.jy.oms.framework.asyn.task.executor.bo.TaskPollResult;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.model.*;
import com.jy.oms.framework.asyn.task.model.strategy.ExclusiveGroupExecuteStrategy;
import com.jy.oms.framework.asyn.task.model.strategy.ExclusiveSameBillExecuteStrategy;
import com.jy.oms.framework.asyn.task.queue.ITaskQueueManager;
import com.jy.oms.framework.asyn.task.queue.TaskQueueManagerProxy;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 默认任务抽取器，适用于【非集群多节点】环境
 *
 * @author liuquan
 */
@Service
public class GroupTaskExtractor implements ITaskExtractor {
    @Autowired
    protected ITaskHandlerManager taskHandlerManager;
    @Autowired
    protected TimeoutTaskCollector timeoutTaskCollector;
    @Autowired
    private List<ITaskExtractorFilter> taskExtractorFilters;

    @Override
    public boolean ableExtract(TaskGroup taskGroup) {
        return !ExecutorContext.getExecutorConfig().isEnableCluster();
    }

    @Override
    public TaskPollResult extract(TaskGroup taskGroup) {
        TaskQueryDto taskQueryDto = this.buildTaskQueryDto(taskGroup);
        return this.collectTask(taskGroup, taskQueryDto);
    }

    protected TaskPollResult collectTask(TaskGroup taskGroup, TaskQueryDto taskQueryDto) {
        //设置任务查询过滤参数
        taskExtractorFilters.stream().forEach(f -> f.fillExtractorParam(taskGroup, taskQueryDto));

        //如果是redis/mem队列，则没有数据时一直堵塞
        ITaskQueueManager taskQueueManager = TaskQueueManagerProxy.locate(taskGroup.getTaskQueueType());
        List<TaskWrapper> taskList = taskQueueManager.poll(taskQueryDto);
        if (CollectionUtils.isEmpty(taskList)) {
            return null;
        }

        TaskPollResult taskPollResult = new TaskPollResult();
        taskPollResult.setTaskSize(taskList.size());
        taskPollResult.setExecutableTasks(taskList);
        //过滤排他任务
        this.filterExclusiveTask(taskList, taskPollResult);
        //过滤超时任务
        this.filterTimeoutTask(taskList, taskPollResult);

        return taskPollResult;
    }

    protected TaskQueryDto buildTaskQueryDto(TaskGroup taskGroup) {
        TaskQueryDto taskQueryDto = new TaskQueryDto();
        taskQueryDto.setTaskGroup(taskGroup.getGroup());
        if (TaskQueueTypeEnum.DB == taskGroup.getTaskQueueType()) {
            taskQueryDto.setTaskSizePerQuery(TaskConstants.DB_QUEUE_TASK_SIZE_PER_QUERY);
        } else {
            taskQueryDto.setTaskSizePerQuery(TaskConstants.REDIS_QUEUE_TASK_SIZE_PER_QUERY);
        }

        return taskQueryDto;
    }

    /**
     * 过滤排他任务，一个时间点同一类任务只有一个可以执行
     */
    protected void filterExclusiveTask(List<TaskWrapper> taskList,
                                       TaskPollResult taskPollResult) {
        if (CollectionUtils.isEmpty(taskList)) {
            return;
        }

        List<TaskWrapper> exclusiveTasks = new ArrayList<>();
        List<String> exclusiveGroups = new ArrayList<>();
        Iterator<TaskWrapper> iterator = taskList.iterator();
        while (iterator.hasNext()) {
            TaskWrapper taskWrapper = iterator.next();
            TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskWrapper.getTaskCode());
            if (taskHandlerWrapper.getExecuteStrategy() instanceof ExclusiveGroupExecuteStrategy) {
                ExclusiveGroupExecuteStrategy strategy = (ExclusiveGroupExecuteStrategy) taskHandlerWrapper.getExecuteStrategy();
                if (exclusiveGroups.contains(strategy.getTaskExclusiveGroup())) {
                    exclusiveTasks.add(taskWrapper);
                    iterator.remove();
                } else {
                    exclusiveGroups.add(strategy.getTaskExclusiveGroup());
                }
            } else if (taskHandlerWrapper.getExecuteStrategy() instanceof ExclusiveSameBillExecuteStrategy) {
                TaskDto taskDto = taskWrapper.getTaskDto();
                if (StringUtils.isNotBlank(taskDto.getBillCode())) {
                    String key = taskDto.getBillCode();
                    if (exclusiveGroups.contains(key)) {
                        exclusiveTasks.add(taskWrapper);
                        iterator.remove();
                    } else {
                        exclusiveGroups.add(key);
                    }
                }
            }
        }

        taskPollResult.setExclusiveTasks(exclusiveTasks);
    }

    /**
     * 过滤超时任务
     */
    protected void filterTimeoutTask(List<TaskWrapper> taskList,
                                     TaskPollResult taskPollResult) {
        if (CollectionUtils.isEmpty(taskList)) {
            return;
        }

        Set<String> timeoutTaskCodes = timeoutTaskCollector.getTimeoutTask();
        if (CollectionUtils.isEmpty(timeoutTaskCodes)) {
            return;
        }

        List<TaskWrapper> timeoutTasks = taskList.stream()
                .filter(task -> timeoutTaskCodes.contains(task.getTaskCode()))
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(timeoutTasks)) {
            return;
        }
        taskPollResult.setTimeoutTasks(timeoutTasks);

        List<TaskWrapper> executableTasks = taskList.stream()
                .filter(task -> !timeoutTaskCodes.contains(task.getTaskCode()))
                .collect(Collectors.toList());
        taskPollResult.setExecutableTasks(executableTasks);
    }
}
